package ltd.flython.dlock.redis;

import ltd.flython.dlock.common.DLock;

import java.util.Collections;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

public class RedisLock implements DLock {

    public static final int DEFAULT_LOCK_TTL = 30000;

    RedisLockHelper factory;
    final String name;
    //获取锁信号
    final Semaphore semaphore = new Semaphore(1);
    volatile Thread headWaiter;

    public RedisLock(RedisLockHelper factory, String name) {
        this.factory = factory;
        this.name = name;
    }

    @Override
    public String getName() {
        return name;
    }

    @Override
    public boolean tryLock() {
        Long ttl = tryAcquire();
        //上锁成功
        if (ttl == null){
            return true;
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        boolean locked = tryLock();
        //上锁成功
        if (locked){return true;}

        long start = System.currentTimeMillis();
        boolean semaphored = semaphore.tryAcquire(time, unit);
        if (!semaphored){return false;}
        //阻塞自己直到被唤醒
        try {
            //检查中断标记
            Thread thread = Thread.currentThread();
            while (!thread.isInterrupted()){
                long pass = System.currentTimeMillis() - start;
                if (pass > time){return false;}
                //再次尝试获取锁，因为前面获取信号量可能阻塞了一段时间
                Long ttl = tryAcquire();
                if (ttl == null){
                    return true;
                }
                //等待
                headWaiter = thread;
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(Math.min(ttl,time-pass)));
                headWaiter = null;
            }
            throw new InterruptedException();
        } finally {
            semaphore.release();
        }
    }

    @Override
    public void lock() {
        boolean locked = tryLock();
        //上锁成功
        if (locked){return;}

        semaphore.acquireUninterruptibly();
        //阻塞自己直到被唤醒
        try {
            //检查中断标记
            Thread thread = Thread.currentThread();
            while (!thread.isInterrupted()){
                //再次尝试获取锁，因为前面获取信号量可能阻塞了一段时间
                Long ttl = tryAcquire();
                if (ttl == null){
                    return;
                }
                //等待
                headWaiter = thread;
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(ttl));
                headWaiter = null;
            }
        } finally {
            semaphore.release();
        }
    }


    @Override
    public void unlockDelayed(long time, TimeUnit unit) {
        String threadId = factory.currentThreadId();
        String holderId = factory.getHolderId(getName());
        if (!threadId.equals(holderId)){ return;}
        factory.stopRenewExpireTask(getName());
        this.doRenewExpire((int) unit.toMillis(time),threadId);
    }

    @Override
    public void forceUnlock() {

//        redisService.delete(getName());
    }

    @Override
    public int unlock() {
        return doUnlock();
    }

    @Override
    public void notifyHead() {
        LockSupport.unpark(headWaiter);
    }

    private Long tryAcquire() {
        String threadId = factory.currentThreadId();
        Long ttl = tryAcquire(DEFAULT_LOCK_TTL, threadId);
        return ttl;
    }


    /**
     * @param leaseMs 默认释放时间
     * @param threadId 线程id
     * @return
     */
    private Long tryAcquire(int leaseMs, String threadId) {
        Long ttl = factory.executeLua(Long.class, "" +
                        "if (redis.call('exists', KEYS[1]) == 0) then\n" +
                        "    redis.call('hset', KEYS[1], ARGV[2], 1);\n" +
                        "    redis.call('pexpire', KEYS[1], ARGV[1]);\n" +
                        "    return nil;\n" +
                        "end;\n" +
                        "if (redis.call('hexists',KEYS[1],ARGV[2]) == 1) then\n" +
                        "    redis.call('hincrby',KEYS[1],ARGV[2],1);\n" +
                        "    redis.call('pexpire', KEYS[1], ARGV[1]);\n" +
                        "    return nil;\n" +
                        "end;\n" +
                        "return redis.call('pttl', KEYS[1]);",
                Collections.singletonList(getName()), leaseMs, threadId);
        if (ttl == null){
            //上锁成功，开始定期刷新任务
            factory.scheduleRenewExpire(this,leaseMs,threadId);
        }
        return ttl;
    }

    private int doUnlock(){
        Long remain = factory.executeLua(Long.class, "" +
                        //如果锁已经不存在， 发布锁释放的消息
                        "if (redis.call('exists', KEYS[1]) == 0) then " +
//                        "   redis.call('publish', KEYS[2], ARGV[3]); " +
                        "   return 0; " +
                        "end;" +
                        //如果释放锁的线程和已存在锁的线程不是同一个线程，返回-1
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 0) then\n" +
                        "    return -1;" +
                        "end\n" +
                        //通过hincrby减少重入度，并刷新过期时间，直到counter <= 0 推送解锁消息
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); \n" +
                        "if (counter > 0) then\n" +
                        "    redis.call('pexpire', KEYS[1], ARGV[1]); \n" +
                        "    return counter;\n" +
                        "else\n" +
                        "    redis.call('del', KEYS[1]); \n" +
//                        "    redis.call('publish', KEYS[2], ARGV[3]); \n" +
                        "    return 0;\n" +
                        "end\n" +
                        "return -1;",
                Collections.singletonList(getName()),
                DEFAULT_LOCK_TTL,factory.currentThreadId(), getName());
        if (remain == 0){
            factory.stopRenewExpireTask(getName());
        }
        return remain.intValue();
    }

    /**
     * @param leaseMs
     * @param threadId
     * @return 是否续期成功
     */
    boolean doRenewExpire(int leaseMs,String threadId){
        Long ret = factory.executeLua(Long.class, "" +
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return 0;",
                Collections.singletonList(getName()),
                leaseMs, threadId);
        return ret.equals(1L);
    }
}
